package com.study.mr.etl;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

/**
 * 1.处理缓冲文件将job中设置的缓冲文件路径获取
 * 2.根据缓存路径再结合输入流把pd.txt的内容写到内存容器中维护
 */
public class EtlMapper extends Mapper<LongWritable,Text,Text,NullWritable> {

    private Text outk=new Text();
    /**
     * 处理mapjoin逻辑
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //获取当前行数据
        String line = value.toString();
        //切割数据
        String[] datas = line.split(" ");
        //遍历集合 将字段长度小于等于11的过滤掉
        for (String data : datas) {
            if(data.length()>11){
                outk.set(data);
                context.write(outk,NullWritable.get());
            }else {
                return;
            }
        }
    }
}
